/*
This software is distributed under the Simplified BSD License:

Copyright (c) 2008, Chris Venter <chris.venter@gmail.com>
All rights reserved.

Redistribution and use in source and binary forms, with or without modification, 
are permitted provided that the following conditions are met:

    * Redistributions of source code must retain the above copyright notice, 
    	this list of conditions and the following disclaimer.
    * Redistributions in binary form must reproduce the above copyright notice, 
    	this list of conditions and the following disclaimer in the documentation 
    	and/or other materials provided with the distribution.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY 
EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES 
OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT 
SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, 
INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED 
TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR 
BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 
CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN 
ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH 
DAMAGE.
*/
#ifndef GENERIK_THREADING_H
#define GENERIK_THREADING_H

/** \file threading.h
This file provides objects for dealing with parallel programming.
*/

#include "threads.h"
#include "macros.h"
#include "exceptions.h"

namespace generik
{
	
	/** Controls a group of threads.
	This class "controls" a pool of threads, in the loosest sense of the word.
	"Control" here simply means it replicates a method call over a group of threads.
	It should be used with caution, especially when dealing with the end-of-life
	of threads.

	If used as intended, it should be safe and can make access to a group of threads
	easier.

	\note No tracking of thread-state is done; for example, when this object goes out of
	scope all references to the threads are lost, and the thread data is destroyed.
	Therefore, if any of the threads are still running at this time they will be in 
	for a nasty surprise.
	*/
	template <typename THREAD>
	struct threadgroup
	{
		std::list<THREAD> _threads;
		
		threadgroup(size_t nthreads = 0, const THREAD& T = THREAD())
		{
			GENERIK_TRACER;
			for (size_t i = 0; i < nthreads; ++i)
			{
				_threads.push_back(T);
			}
			std::cout<<"created "<<_threads.size()<<" threads"<<std::endl;
		}
		~threadgroup()
		{
			GENERIK_TRACER;
		}
		
		void execute()
		{
			GENERIK_TRACER;
			GENERIK_FOR_EACH(i, _threads)
			{
				i->execute();
			}
		}
		void cancel()
		{
			GENERIK_TRACER;
			GENERIK_FOR_EACH(i, _threads)
			{
				i->cancel();
			}
		}
		void detach()
		{
			GENERIK_TRACER;
			GENERIK_FOR_EACH(i, _threads)
			{
				i->detach();
			}
		}
		void join()
		{
			GENERIK_TRACER;
			GENERIK_FOR_EACH(i, _threads)
			{
				i->join();
			}
		}
		void stop()
		{
			GENERIK_TRACER;
			GENERIK_FOR_EACH(i, _threads)
			{
				i->stop();
			}
		}
	};
	
	template <typename THREAD>
	struct threadgroup<THREAD*>
	{
		std::list<THREAD*> _threads;
		
		void execute()
		{
			GENERIK_TRACER;
			GENERIK_FOR_EACH(i, _threads)
			{
				(*i)->execute();
			}
		}
		void cancel()
		{
			GENERIK_TRACER;
			GENERIK_FOR_EACH(i, _threads)
			{
				(*i)->cancel();
			}
		}
		void detach()
		{
			GENERIK_TRACER;
			GENERIK_FOR_EACH(i, _threads)
			{
				(*i)->detach();
			}
		}
		void join()
		{
			GENERIK_TRACER;
			GENERIK_FOR_EACH(i, _threads)
			{
				(*i)->join();
			}
		}
		void stop()
		{
			GENERIK_TRACER;
			GENERIK_FOR_EACH(i, _threads)
			{
				(*i)->stop();
			}
		}
	};
	
	template <typename INPUT, typename OUTPUT>
	struct threaddata
	{
		virtual INPUT pop() = 0;
		virtual void push(OUTPUT& out) = 0;
		virtual bool empty() = 0;
		
		INPUT pop_()
		{
			wait_for_input_available();
			// _read is now locked, and input is available
			INPUT in = pop();
			++_in_counter;
			_read.unlock();
			return in;
		}
		void push_(OUTPUT& out)
		{
			_write.lock();
			push(out);
			++_out_counter;
			_done.signal_locked(); // unlocks _write!
		}
		
		/// allows access to the input data
		generik::mutex _read;
		/// allows access to the output data (and _busy counter)
		generik::mutex _write;
		
		/// this uses _read to get access to the input data, and determine when it is empty
		/// it accompanies the check [empty()]
		generik::condition _input_empty;
		/// this uses _read to get access to the input data, and determine when there is available data
		/// it accompanies the check [!empty()]
		generik::condition _input_available;
		
		/// this is the counter of the number of input data items received
		size_t _in_counter;
		/// this is the counter of the number of output data items calculated
		size_t _out_counter;
		/// this uses _write, and determines when a thread is done
		/// the check it accompanies [is _in_counter == _out_counter]
		generik::condition _done;
		
		threaddata()
		: _read("read")
		, _write("write")
		, _input_empty(_read)
		, _input_available(_read)
		, _in_counter(0), _out_counter(0)
		, _done(_write)
		{
			GENERIK_TRACER;
		}
		
		/*
		This gets a read lock when we know there is input available,
		so we can safely read without anyone else interfering.
		*/
		void wait_for_input_available()
		{
			GENERIK_TRACER;
			_read.lock(); // obtain the lock
			if (empty()) // check if there is any data
			{
				_read.unlock();
				// there is no data, so signal any writers to produce more data
				_input_empty.signal();
				// ...and wait until they do so
				_read.lock(); while (empty()) _input_available.wait_locked();
			}
			// _read is now locked, and _input should(!) have data
			GENERIK_ASSERT(!empty());
		}
		
		/*
		This gets a read lock when we know there is NO input available,
		so we can safely add input data without anyone else interfering.
		*/
		void wait_for_input_empty()
		{
			GENERIK_TRACER;
			_read.lock(); // obtain the initial lock
			while (!empty()) // check whether there is in fact no data
			{
				_read.unlock();
				// there is data, so signal any readers to do their job
				_input_available.broadcast();
				// ... and wait until they do so
				_read.lock();
				if (!empty()) _input_empty.wait_locked();
			}
			// _read is now locked, and _input should(!) be empty
			GENERIK_ASSERT(empty());
		}
		
		/*
		This gets a read lock with no input available (ie. left over)
		and then gets a write lock once all input data has been accounted for
		(ie. _in_counter == _out_counter). 
		This means the calculation should be over!
		*/
		void wait_for_idle()
		{
			GENERIK_TRACER;
			wait_for_input_empty();
			// _read is now locked
			_write.lock();
			while (_in_counter > _out_counter) _done.wait_locked();
			// _write is now locked, and _in_counter <= _out_counter!
			// (well, at least _in_counter == _out_counter, since we can't produce more than we consume?
			GENERIK_TRACE_VALUE(_in_counter);
			GENERIK_TRACE_VALUE(_out_counter);
			GENERIK_ASSERT(_in_counter == _out_counter);
		}
		
		void unlock_all()
		{
			GENERIK_TRACER;
			_read.unlock();
			_write.unlock();
		}
	};
	
	/** Interface to a threadable task
	*/
	struct threadtask
	{
		virtual void* run() = 0;
	};
	
	/** A worker class.
	This thread-derived class holds a reference to a threadtask, runs the task
	as its main execution function.
	*/
	struct worker
	: thread
	{
		threadtask* _task;
		
		worker(threadtask* task_in)
		: _task(task_in)
		{
			GENERIK_TRACER;
		}
		
		void* run()
		{
			GENERIK_TRACER;
			return _task->run();
		}
		
	};
	
	/** Manages a threadpool which accepts INPUT data and produces OUTPUT data.
	It is a base class that manages a pool of threads on an abstracted set of
	input and output data. It implements the control over the threads with regards
	to access to the data, as well as the thread life-time.
	
	In order to execute a task using a threadpool, derive from this class, implmenting
	the input- and output-container interface functions, as well as the process
	function.
	*/
	template <typename INPUT, typename OUTPUT>
	struct threadpool
	: threadtask
	, threaddata<INPUT, OUTPUT>
	, threadgroup<worker>
	{
		/** the derived implementation class will specialise
		this to use pop_ and push_ to retrieve and store data,
		processing it along the way.
		*/
		virtual OUTPUT process(const INPUT& input_in) = 0;
		
		void* final;
		/// this is run by each thread in the pool
		void* run()
		{
			GENERIK_TRACER;
			final = 0;
			while (!final)
			{
				INPUT in = this->pop_();
				OUTPUT out = process(in);
				this->push_(out);
			}
			return final;
		}
		
		threadpool(size_t nthreads = 0)
		: threadgroup<worker>(nthreads, this)
		{
			GENERIK_TRACER;
			GENERIK_ASSERT(nthreads > 0);
		}
		
		/**
		\note when using finish() directly after execute(), it may be necessary
		to initialise the input data store beforehand.
		Since finish() locks _read AND _write (by using wait_for_idle) the situation
		MAY occur where a supplier may not have had a chance to add input data
		(ie. leaving _in_counter at zero) by the time we compare it to _out_counter,
		therefore leaving the calculation before it has begun!
		Therefore some thought is required before naively writing
			execute();
			finish();
		*/
		void finish()
		{
			GENERIK_TRACER;
			this->wait_for_idle();
			this->unlock_all();
			stop();
		}
	};
	
	template <typename INPUT>
	struct slave;
	
	template <typename INPUT>
	struct slavetask
	{
		virtual void* run(const INPUT& input_in, slave<INPUT>* slave_in) = 0;
	};
	
	template <typename INPUT>
	struct slave
	: thread
	{
		slavetask<INPUT>* _task;
		INPUT _input;
		
		slave(slavetask<INPUT>* task_in, const INPUT& input_in)
		: _task(task_in)
		, _input(input_in)
		{
			GENERIK_TRACER;
		}
		
		void* run()
		{
			GENERIK_TRACER;
			return _task->run(_input, this);
		}
	};
	
	template <typename INPUT, typename OUTPUT>
	struct threadgenerator
	: threaddata<INPUT, OUTPUT>
	, slavetask<INPUT>
	, threadgroup<slave<INPUT>*>
	, thread
	{
		size_t _running_threads;
		
		mutex _access;
		mutex _shutdown;
		bool _finished;
		condition _available;
		size_t _max_threads;
		
		threadgenerator(size_t max_threads)
		: _running_threads(0)
		, _access("access")
		, _shutdown("shutdown")
		, _finished(false)
		, _available(_access)
		, _max_threads(max_threads)
		{
			GENERIK_TRACER;
			GENERIK_ASSERT(max_threads > 0);
		}
		
		bool available()
		{
			return this->_threads.size() < _max_threads;
		}
		
		void* final;
		/// this is run by the generator itself
		void* run()
		{
			GENERIK_TRACER;
			final = 0;
			while (!final)
			{
				INPUT in = this->pop_();
				_access.lock();
				while (!available()) _available.wait_locked();
				GENERIK_ASSERT(available());
				// and _access is locked!
				GENERIK_TRACE("threading: generating new thread "<<++_running_threads);
				slave<INPUT>* s = new slave<INPUT>(this, in);
				this->_threads.push_back(s);
				s->execute();
				_access.unlock();
			}
			return final;
		}
		
		virtual OUTPUT process(const INPUT& input_in) = 0;
		
		/// this is run by each generated thread
		void* run(const INPUT& input_in, slave<INPUT>* slave_in)
		{
			GENERIK_TRACER;
			final = 0;
			OUTPUT out = process(input_in);
			this->push_(out);
			//
			_shutdown.lock();
			if (!_finished)
			{
				_access.lock();
				GENERIK_TRACE("threading: exiting thread (zombie) "<<--_running_threads);
				_shutdown.unlock();
				slave_in->detach();
				this->_threads.remove(slave_in);
				delete slave_in;
				_available.signal_locked(); // unlocks _access!
			}
			else
			{
				GENERIK_TRACE("threading: exiting thread "<<--_running_threads);
				_shutdown.unlock();
			}
			return final;
		}
		
		void execute()
		{
			GENERIK_TRACER;
			thread::execute();
		}
		
		void stop()
		{
			GENERIK_TRACER;
			thread::stop();
			_shutdown.lock();
			_finished = true;
			_access.lock();
			_shutdown.unlock();
			threadgroup<slave<INPUT>*>::stop();
			_access.unlock();
		}
		
		void finish()
		{
			GENERIK_TRACER;
			this->wait_for_idle();
			this->unlock_all();
			stop();
		}
		
	};
	
}

#endif


